import time
import asyncio

"""
# 同步版本
def sync_demo():
    def task(name, delay):
        print(f"{name} 开始")
        time.sleep(delay)  # 阻塞调用
        print(f"{name} 完成")

    start = time.time()
    task("任务1", 2)
    task("任务2", 1)
    print(f"总耗时: {time.time() - start:.2f}秒")


# 异步版本
async def async_demo():
    async def task(name, delay):
        print(f"{name} 开始")
        await asyncio.sleep(delay)  # 非阻塞等待
        print(f"{name} 完成")

    start = time.time()
    await asyncio.gather(task("任务1", 2), task("任务2", 1))
    print(f"总耗时: {time.time() - start:.2f}秒")


# 运行对比
print("=== 同步版本 ===")
sync_demo()  # 耗时约3秒

print("\n=== 异步版本 ===")
# 任务一和任务同时开始，以最长耗时任务结束为准
asyncio.run(async_demo())  # 耗时约2秒


async def my_coroutine():
    print("协程开始执行")
    await asyncio.sleep(1)
    print("协程结束")
    return "协程结果"


print(asyncio.run(my_coroutine()))


async def simple_coroutine():
    print("协程开始")
    await asyncio.sleep(1)
    print("协程结束")
    return "协程结果"


async def coroutine_usage():
    # 1. 直接等待
    print("=== 直接等待 ===")
    result = await simple_coroutine()
    print(f"结果: {result}")

    # 2. 创建任务
    print("\n=== 创建任务 ===")
    task = asyncio.create_task(simple_coroutine())
    result = await task
    print(f"任务结果: {result}")

    # 3. 并发执行
    print("\n=== 并发执行 ===")
    tasks = [simple_coroutine(), simple_coroutine(), simple_coroutine()]
    results = await asyncio.gather(*tasks)
    print(f"所有结果: {results}")


asyncio.run(coroutine_usage())

# 获取事件循环
loop = asyncio.get_event_loop()
print(f"事件循环: {loop}")
print(f"循环是否运行: {loop.is_running()}")
print(f"循环是否关闭: {loop.is_closed()}")


# 手动管理事件循环
async def simple_task():
    await asyncio.sleep(1)
    return "任务完成"


# 方式1：手动管理
loop = asyncio.get_event_loop()
try:
    if not loop.is_running():
        result = loop.run_until_complete(simple_task())
        print(f"手动执行结果: {result}")
finally:
    loop.close()

# 方式2：使用 asyncio.run() (推荐)
result = asyncio.run(simple_task())
print(result)
"""

# class AsyncExamples:
#     """异步编程示例"""

#     async def io_operation(self, name, delay):
#         """模拟I/O操作"""
#         print(f"{name}: 开始I/O操作，需要{delay}秒")
#         await asyncio.sleep(delay)
#         print(f"{name}: I/O操作完成")
#         return f"{name}_result"

#     async def cpu_operation(self, name, iterations):
#         """模拟CPU密集型操作"""
#         print(f"{name}: 开始CPU计算")
#         result = 0
#         for i in range(iterations):
#             result += i
#         print(f"{name}: CPU计算完成")
#         return result

#     async def concurrent_operations(self):
#         """并发操作示例"""
#         tasks = [self.io_operation(f"任务{i}", i) for i in range(1, 4)]

#         print("=== 并发I/O操作 ===")
#         results = await asyncio.gather(*tasks)
#         print(f"并发结果: {results}")


# # 使用示例
# examples = AsyncExamples()
# asyncio.run(examples.concurrent_operations())


# class AsyncDatabaseConnection:
#     async def __aenter__(self):
#         print("建立数据库连接...")
#         await asyncio.sleep(0.5)
#         print("数据库连接已建立")
#         return self

#     async def __aexit__(self, exc_type, exc_val, exc_tb):
#         print("关闭数据库连接...")
#         await asyncio.sleep(0.2)
#         print("数据库连接已关闭")
#         return False

#     async def execute_query(self, query):
#         print(f"执行查询: {query}")
#         await asyncio.sleep(0.3)
#         return f"查询结果: {query}"


# async def async_context():
#     async with AsyncDatabaseConnection() as db:
#         result = await db.execute_query("SELECT * FROM users")
#         print(result)


# asyncio.run(async_context())


# class AsyncDataStream:
#     """异步数据流"""

#     def __init__(self, data_list, delay=0.5):
#         self.data = data_list
#         self.delay = delay
#         self.index = 0

#     def __aiter__(self):
#         return self

#     async def __anext__(self):
#         if self.index >= len(self.data):
#             raise StopAsyncIteration

#         item = self.data[self.index]
#         self.index += 1

#         await asyncio.sleep(self.delay)
#         print(f"生成数据: {item}")
#         return item


# async def async_iterator():
#     """异步迭代器演示"""

#     stream = AsyncDataStream(["数据1", "数据2", "数据3", "数据4"], 0.5)

#     print("使用异步for循环:")
#     async for item in stream:
#         print(f"接收到: {item}")

#     print("\n使用异步推导式:")
#     stream2 = AsyncDataStream(["A", "B", "C", "D"], 1.2)
#     results = [item async for item in stream2]
#     print(f"所有数据: {results}")


# asyncio.run(async_iterator())

# import asyncio
# import logging

# logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(message)s")


# class TaskManager:
#     """任务管理器"""

#     def __init__(self):
#         self.tasks = set()

#     async def worker(self, name, duration):
#         """工作协程"""
#         logging.info(f"任务 {name} 开始")
#         try:
#             await asyncio.sleep(duration)
#             logging.info(f"任务 {name} 完成")
#             return f"{name}_result"
#         except asyncio.CancelledError:
#             logging.info(f"任务 {name} 被取消")
#             raise

#     async def create_tasks(self, count):
#         """创建多个任务"""
#         for i in range(count):
#             task = asyncio.create_task(
#                 self.worker(f"worker_{i}", i + 1), name=f"task_{i}"
#             )
#             self.tasks.add(task)
#             task.add_done_callback(self.tasks.discard)

#         logging.info(f"创建了 {count} 个任务")

#     async def monitor_tasks(self):
#         """监控任务状态"""
#         while True:
#             running_tasks = [t for t in self.tasks if not t.done()]
#             logging.info(f"运行中的任务: {len(running_tasks)}")

#             if not running_tasks:
#                 break

#             await asyncio.sleep(0.5)

#     async def cancel_all_tasks(self):
#         """取消所有任务"""
#         logging.info("取消所有任务")
#         for task in self.tasks:
#             task.cancel()

#         await asyncio.gather(*self.tasks, return_exceptions=True)


# async def task_management():
#     """任务管理演示"""
#     manager = TaskManager()

#     await manager.create_tasks(5)
#     monitor_task = asyncio.create_task(manager.monitor_tasks())

#     await asyncio.sleep(2)
#     await manager.cancel_all_tasks()

#     await monitor_task


# asyncio.run(task_management())
# import asyncio


# async def future_example():
#     """Future 对象示例"""

#     # 创建一个Future对象
#     future = asyncio.Future()
#     print(f"Future初始状态: {future.done()}")

#     async def set_future_result():
#         await asyncio.sleep(1)
#         if not future.done():
#             future.set_result("Future结果")
#             print("Future结果已设置")

#     async def use_future():
#         print("等待Future结果...")
#         result = await future
#         print(f"收到Future结果: {result}")
#         return result

#     # 并发执行设置和获取Future结果
#     results = await asyncio.gather(set_future_result(), use_future())

#     print(f"所有操作完成: {results}")


# asyncio.run(future_example())
# import asyncio
# import random


# class AsyncProducerConsumer:
#     """异步生产者-消费者"""

#     def __init__(self, queue_size=5):
#         self.queue = asyncio.Queue(maxsize=queue_size)
#         self.producers = []
#         self.consumers = []

#     async def producer(self, name, count):
#         """生产者"""
#         for i in range(count):
#             item = f"{name}_item_{i}"
#             production_time = random.uniform(0.1, 0.5)
#             await asyncio.sleep(production_time)

#             await self.queue.put(item)
#             print(f"生产者 {name} 生产了: {item} (队列大小: {self.queue.qsize()})")

#         print(f"生产者 {name} 完成")

#     async def consumer(self, name):
#         """消费者"""
#         while True:
#             try:
#                 item = await asyncio.wait_for(self.queue.get(), timeout=2.0)

#                 consumption_time = random.uniform(0.2, 0.8)
#                 await asyncio.sleep(consumption_time)

#                 print(f"消费者 {name} 消费了: {item} (队列大小: {self.queue.qsize()})")
#                 self.queue.task_done()

#             except asyncio.TimeoutError:
#                 print(f"消费者 {name} 超时，退出")
#                 break

#     async def run(self, producer_count=2, items_per_producer=5, consumer_count=3):
#         """运行生产者-消费者系统"""
#         print("启动生产者-消费者系统...")

#         # 创建生产者任务
#         for i in range(producer_count):
#             producer_task = asyncio.create_task(
#                 self.producer(f"P{i}", items_per_producer)
#             )
#             self.producers.append(producer_task)

#         # 创建消费者任务
#         for i in range(consumer_count):
#             consumer_task = asyncio.create_task(self.consumer(f"C{i}"))
#             self.consumers.append(consumer_task)

#         # 等待所有生产者完成
#         await asyncio.gather(*self.producers)
#         print("所有生产者已完成")

#         # 等待队列中所有项目被消费
#         await self.queue.join()
#         print("队列已清空")

#         # 取消所有消费者任务
#         for consumer in self.consumers:
#             consumer.cancel()

#         await asyncio.gather(*self.consumers, return_exceptions=True)
#         print("所有消费者已结束")


# # 使用示例
# system = AsyncProducerConsumer()
# asyncio.run(system.run())

# import asyncio

# # 基本用法
# lock = asyncio.Lock()


# async def task_with_lock():
#     async with lock:
#         # 此处代码同一时刻只能有一个协程执行
#         print("task start")
#         await asyncio.sleep(1)
#         print("task end")


# async def run():
#     # await asyncio.gather(task_with_lock(), task_with_lock(), task_with_lock())
#     tasks = [task_with_lock() for _ in range(3)]
#     await asyncio.gather(*tasks)


# asyncio.run(run())

# import asyncio

# # 创建最大允许2个协程同时运行的信号量
# sem = asyncio.Semaphore(2)


# async def limited_task():
#     async with sem:
#         print("task start")
#         await asyncio.sleep(1)
#         print("task end")


# async def main():
#     tasks = [limited_task() for _ in range(5)]
#     await asyncio.gather(*tasks)


# asyncio.run(main())
# import asyncio


# class AsyncResourceManager:
#     """异步资源管理"""

#     def __init__(self, max_concurrent=3):
#         self.semaphore = asyncio.Semaphore(max_concurrent)
#         self.lock = asyncio.Lock()
#         self.shared_resource = 0

#     async def access_limited_resource(self, name):
#         """访问受限制的资源"""
#         async with self.semaphore:
#             print(f"{name} 获得资源访问权")
#             await asyncio.sleep(1)  # 模拟资源使用
#             print(f"{name} 释放资源访问权")

#     async def update_shared_resource(self, name, value):
#         """更新共享资源（需要锁）"""
#         async with self.lock:
#             print(f"{name} 获得锁，当前资源值: {self.shared_resource}")
#             await asyncio.sleep(0.5)  # 模拟处理时间
#             self.shared_resource += value
#             print(f"{name} 更新资源为: {self.shared_resource}")
#             print(f"{name} 释放锁")


# async def concurrency_control():
#     """并发控制演示"""
#     manager = AsyncResourceManager(max_concurrent=2)

#     # 测试信号量
#     print("=== 信号量测试 ===")
#     semaphore_tasks = [manager.access_limited_resource(f"任务{i}") for i in range(5)]
#     await asyncio.gather(*semaphore_tasks)

#     # 测试锁
#     print("\n=== 锁测试 ===")
#     lock_tasks = [manager.update_shared_resource(f"更新者{i}", i + 1) for i in range(3)]
#     await asyncio.gather(*lock_tasks)


# asyncio.run(concurrency_control())
# 导入asyncio模块，用于异步编程
# import asyncio

# # 导入aiohttp模块，用于异步HTTP请求
# import aiohttp

# # 导入time模块，用于计时
# import time


# # 定义异步Web客户端类
# class AsyncWebClient:
#     # 声明类的用途为异步Web客户端
#     """异步Web客户端"""

#     # 初始化方法
#     def __init__(self):
#         # 初始化session为None
#         self.session = None

#     # 定义异步上下文管理器进入方法
#     async def __aenter__(self):
#         # 创建aiohttp客户端会话
#         self.session = aiohttp.ClientSession()
#         # 返回自身
#         return self

#     # 定义异步上下文管理器退出方法
#     async def __aexit__(self, exc_type, exc_val, exc_tb):
#         # 关闭aiohttp客户端会话
#         await self.session.close()

#     # 定义异步方法，用于获取指定URL的内容
#     async def fetch_url(self, url, name):
#         # 获取URL内容
#         """获取URL内容"""
#         # 打印请求开始信息
#         print(f"{name}: 开始请求 {url}")

#         try:
#             # 以异步方式发起GET请求，10秒超时
#             async with self.session.get(url, timeout=10) as response:
#                 # 读取响应的文本内容
#                 content = await response.text()
#                 # 打印请求完成信息，包括状态码和内容长度
#                 print(f"{name}: 完成，状态码 {response.status}, 长度 {len(content)}")
#                 # 返回内容长度
#                 return len(content)
#         except Exception as e:
#             # 捕获异常并打印错误信息
#             print(f"{name}: 错误 {e}")
#             # 出错时返回0
#             return 0

#     # 定义异步方法，用于并发请求多个URL
#     async def concurrent_requests(self, urls):
#         # 并发请求多个URL
#         """并发请求多个URL"""
#         # 创建任务列表
#         tasks = []

#         # 遍历所有URL，创建对应的fetch_url任务
#         for i, url in enumerate(urls):
#             # 创建单个请求任务
#             task = self.fetch_url(url, f"请求{i+1}")
#             # 添加任务到任务列表
#             tasks.append(task)

#         # 打印并发请求开始信息
#         print("开始并发请求...")
#         # 记录开始时间
#         start_time = time.time()

#         # 并发执行所有异步请求，等待结果
#         results = await asyncio.gather(*tasks)

#         # 记录结束时间
#         end_time = time.time()
#         # 打印所有请求完成和耗时信息
#         print(f"所有请求完成，耗时: {end_time - start_time:.2f}秒")
#         # 打印所有任务的结果
#         print(f"结果: {results}")

#         # 返回结果列表
#         return results


# # 定义异步函数，演示Web客户端的用法
# async def web_client():
#     # Web客户端演示
#     """Web客户端演示"""
#     # 准备待请求的URL列表
#     urls = [
#         "https://httpbin.org/delay/1",
#         "https://httpbin.org/delay/2",
#         "https://httpbin.org/delay/1",
#         "https://httpbin.org/status/200",
#         "https://httpbin.org/status/404",
#     ]

#     # 使用AsyncWebClient异步上下文管理器
#     async with AsyncWebClient() as client:
#         # 并发请求所有URL
#         await client.concurrent_requests(urls)


# # 运行Web客户端演示
# asyncio.run(web_client())
# 导入异步相关库
# import asyncio

# # 导入时间库用于计时
# import time

# # 导入requests库用于同步HTTP请求
# import requests

# # 导入aiohttp库用于异步HTTP请求
# import aiohttp


# # 定义性能比较器类
# class PerformanceComparator:
#     # 类的说明文档
#     """性能比较器"""

#     # 定义静态方法用于异步HTTP请求
#     @staticmethod
#     async def async_http_requests(urls):
#         # 方法的说明文档
#         """异步HTTP请求"""
#         # 创建异步HTTP会话
#         async with aiohttp.ClientSession() as session:
#             # 初始化任务列表
#             tasks = []
#             # 遍历每一个url
#             for url in urls:
#                 # 创建GET请求任务
#                 task = session.get(url)
#                 # 添加任务到任务列表
#                 tasks.append(task)
#             # 并发执行所有任务并获取响应
#             responses = await asyncio.gather(*tasks)
#             # 返回所有响应的状态码
#             return [resp.status for resp in responses]

#     # 定义静态方法用于同步HTTP请求
#     @staticmethod
#     def sync_http_requests(urls):
#         # 方法的说明文档
#         """同步HTTP请求"""
#         # 初始化结果列表
#         results = []
#         # 遍历每一个url
#         for url in urls:
#             # 发送GET请求
#             response = requests.get(url)
#             # 添加响应状态码到结果列表
#             results.append(response.status_code)
#         # 返回所有结果
#         return results

#     # 定义异步方法用于比较性能
#     async def compare_performance(self, urls):
#         # 方法的说明文档
#         """比较性能"""
#         # 打印待测试的URL数量
#         print(f"测试 {len(urls)} 个URL")

#         # --- 同步部分 ---
#         # 打印同步测试标题
#         print("=== 同步版本 ===")
#         # 记录开始时间
#         start = time.time()
#         # 进行同步HTTP请求
#         sync_results = self.sync_http_requests(urls)
#         # 计算同步请求耗时
#         sync_time = time.time() - start
#         # 打印同步请求耗时
#         print(f"同步耗时: {sync_time:.2f}秒")

#         # --- 异步部分 ---
#         # 打印异步测试标题
#         print("\n=== 异步版本 ===")
#         # 记录开始时间
#         start = time.time()
#         # 进行异步HTTP请求
#         async_results = await self.async_http_requests(urls)
#         # 计算异步请求耗时
#         async_time = time.time() - start
#         # 打印异步请求耗时
#         print(f"异步耗时: {async_time:.2f}秒")

#         # 计算性能提升倍数
#         improvement = sync_time / async_time
#         # 打印性能提升结果
#         print(f"\n性能提升: {improvement:.2f}x")

#         # 返回同步和异步的结果
#         return sync_results, async_results


# # 定义异步性能演示函数
# async def performance():
#     # 方法文档说明
#     """性能演示"""
#     # 使用httpbin.org作为测试URL，重复5次
#     urls = ["https://httpbin.org/delay/1"] * 5

#     # 创建性能比较器对象
#     comparator = PerformanceComparator()
#     # 调用性能比较方法并等待结果
#     await comparator.compare_performance(urls)


# # 启动性能测试
# asyncio.run(performance())
